# -*- coding:utf-8 -*-

__author__ = 'ghostviper'

import sys
sys.path.append("..")
from kafka import KafkaConsumer,KafkaProducer
from kafka.common import KafkaError
from kafka.common import OffsetOutOfRangeError
from collections import OrderedDict
from kafka.common import KafkaUnavailableError
from method_timer import MethodTimer
from retrying import retry
import traceback
import time
import json


class KafkaTool(object):

    consumer = None

    def __init__(self, settings_name):
        """
        设置kafka配置文件
        :param settings_name:
        """
        self.settings = settings_name

    def _setup_kafka(self):
        """
        启动一个kafka链接
        :return:
        """
        self.consumer = self._create_consumer()
        print "Successfully connected Kafka"

    @retry(wait_exponential_multiplier=500, wait_exponential_max=10000)
    def _create_consumer(self):
        """
        创建一个kafka消费链接
        :return:
        """
        try:
            brokers = self.settings['KAFKA_HOSTS']
            print "create new kafka consumer using brokers: %s and topic %s" % (self.settings['KAFKA_HOSTS'],
                                                                                self.settings['KAFKA_INCOMING_TOPIC'])
            return KafkaConsumer(
                    self.settings['KAFKA_INCOMING_TOPIC'],
                    group_id=self.settings['KAFKA_GROUP'],
                    bootstrap_servers=brokers,
                    consumer_timeout_ms=self.settings['KAFKA_CONSUMER_TIMEOUT'],
                    auto_offset_reset=self.settings['KAFKA_CONSUMER_AUTO_OFFSET_RESET'],
                    auto_commit_interval_ms=self.settings['KAFKA_CONSUMER_COMMIT_INTERVAL_MS'],
                    enable_auto_commit=self.settings['KAFKA_CONSUMER_AUTO_COMMIT_ENABLE'],
                    max_partition_fetch_bytes=self.settings['KAFKA_CONSUMER_FETCH_MESSAGE_MAX_BYTES']
            )
        except KeyError as e:
            print "missing setting named ", str(e), {'ex': traceback.format_exc()}
        except:
            print "couldn't init kafka consumer for topic", {'ex': traceback.format_exc()}
            raise

    def _create_producer(self):
        """
        创建一个kafka生产者链接
        :return:
        """
        try:
            brokers = self.settings['KAFKA_HOSTS']
            print "create new kafka producer using brokers: ", str(brokers)
            return KafkaProducer(
                    bootstrap_servers=brokers,
                    value_serializer=lambda m: json.dumps(m),
                    retries=3,
                    linger_ms=self.settings['KAFKA_PRODUCER_BATCH_LINGER_MS'],
                    buffer_memory=self.settings['KAFKA_PRODUCER_BUFFER_BYTES']
            )
        except KeyError as e:
            print "missing setting named ", str(e), {'e': traceback.format_exc()}
        except:
            print "couldn't init kafka producer.", {'ex': traceback.format_exc()}
            raise

    def _main_loop(self):
        while True:
            try:
                for message in self.consumer:
                    if message is None:
                        print "no message"
                    print message
            except OffsetOutOfRangeError:
                self.consumer.seek_to_end()
                print "kafka consuming complete!"
            except:
                print "kafka consuming failed!"
                raise
            time.sleep(5)


    def feed(self, json_item):
        """
        向一个kafka topic喂一条json信息
        :param json_item:
        :return:
        """
        @MethodTimer.timeout(self.settings['KAFKA_FEED_TIMEOUT'], False)
        def _feed(json_item):
            producer = self._create_producer()
            topic = self.settings['KAFKA_INCOMING_TOPIC']
            if producer is not None:
                producer.send(topic, json_item)
                producer.flush()
                producer.close()
                return True
            else:
                return False
        result = _feed(json_item)
        if result:
            print "successfully feed item to kafka"
        else:
            print "failed to feed item into kafka"

    def run(self):
        '''
        Set up and run
        '''
        self._setup_kafka()
        self._main_loop()

    def close(self):
        if self.consumer is not None:
            self.consumer.close()




